package com.etlmaster.executor.bigdata.hbase;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;

import com.etlmaster.executor.utils.LogWriter;



/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
public class HbaseUtil {
	private static Configuration conf = HBaseConfiguration.create();

	static {
		addResource(Environment.CORE_SITE);
		addResource(Environment.HDFS_SITE);
		addResource(Environment.HBSE_SITE);
	}

	private static void addResource(String filePath) {
		File file = new File(filePath);
		if (file.exists()) {
			Path path = new Path(filePath);
			conf.addResource(path);
		} else {
			LogWriter.addLog("WARN", "配置文件{}不存在！请检查", filePath);
		}
	}

	/**
	 * 
	 * @description 在Hbase中创建表
	 * 
	 * @param tableName
	 * 
	 *            表名
	 * 
	 * @param cfs
	 * 
	 *            列族
	 * 
	 * @throws IOException
	 */
	public static void create(HBaseAdmin admin, String tableName, String[] cfs,
			boolean isOverride) throws IOException {

		if (admin.tableExists(tableName)) {
			if (isOverride) {
				LogWriter
						.addLog("warn", "表[" + tableName + "]已经存在！！！删除原表，重新创建");
				delete(admin, tableName);
				create(admin, tableName, cfs);
			} else {
				LogWriter.addLog("warn", "表[" + tableName + "]已经存在！！！");
			}
		} else {
			create(admin, tableName, cfs);
		}
		closeAdmin(admin);

	}

	public static void delete(HBaseAdmin admin, String tableName)
			throws IOException {
		admin.disableTable(tableName);
		admin.deleteTable(tableName);
	}

	public static void create(HBaseAdmin admin, String tableName, String[] cfs)
			throws IOException {
		TableName tableNameObj = TableName.valueOf(tableName);
		HTableDescriptor tableDesc = new HTableDescriptor(tableNameObj);
		for (int i = 0; i < cfs.length; i++) {
			tableDesc.addFamily(new HColumnDescriptor(cfs[i]));
		}
		admin.createTable(tableDesc);
		LogWriter.addLog("info", "表[" + tableName + "]创建成功！！！");
	}

	public static void create(Configuration conf, String tableName, String[] cfs)
			throws IOException {
		HBaseAdmin admin = null;

		admin = new HBaseAdmin(conf);
		if (admin.tableExists(tableName)) {
			LogWriter.addLog("warn", "表[" + tableName + "]已经存在！！！");

		} else {
			TableName tableNameObj = TableName.valueOf(tableName);
			HTableDescriptor tableDesc = new HTableDescriptor(tableNameObj);
			for (int i = 0; i < cfs.length; i++) {
				tableDesc.addFamily(new HColumnDescriptor(cfs[i]));
			}
			admin.createTable(tableDesc);
			LogWriter.addLog("info", "表[" + tableName + "]创建成功！！！");
		}

	}

	public static void put(String tableName, String rowKey, String cf,
			String key, String value) {
		HTable table = null;
		try {
			table = new HTable(getConf(), tableName);
			Put put = new Put(Bytes.toBytes(rowKey));
			put.add(Bytes.toBytes(cf), Bytes.toBytes(key), Bytes.toBytes(value));
			table.put(put);
		} catch (IOException e) {
			LogWriter.addLog("error", "put操作失败！！！{}", e.getMessage());
		} finally {
			closeTable(table);
		}
	}

	protected static String getRowKey(Map<String, Object> row,
			String[] pkColumnNames) {
		StringBuilder str = new StringBuilder();
		for (String columnName : pkColumnNames) {
			str.append(row.get(columnName));
		}
		return str.toString();
	}

	public static void batchPut(Configuration conf, String hTableName,
			String rowKeyColumn, List<Map<String, Object>> rows) {
		HTable table = null;
		try {

			table = new HTable(conf, hTableName);
			List<Put> puts = new ArrayList<Put>();
			int commitCount = 10000;// 每次最多提交的行数

			for (Map<String, Object> row : rows) {
				String[] rowKeyColumns = StringUtils
						.splitByWholeSeparatorPreserveAllTokens(rowKeyColumn,
								",");
				Iterator<String> columnNames = row.keySet().iterator();
				StringBuilder builder = new StringBuilder();
				String rowKey = null;
				while (columnNames.hasNext()) {
					String columnName = columnNames.next();
					for (String column : rowKeyColumns) {
						if (columnName.equalsIgnoreCase(column)) {
							builder.append(String.valueOf(row.get(columnName)));
						}
					}
				}
				rowKey = builder.toString();
				Put put = new Put(Bytes.toBytes(rowKey));
				Iterator<String> columnNames2 = row.keySet().iterator();
				while (columnNames2.hasNext()) {
					String columnName = columnNames2.next();
					String columnValue = String.valueOf(row.get(columnName));
					put.add(Bytes.toBytes(columnName), Bytes.toBytes(""),
							Bytes.toBytes(columnValue));
					puts.add(put);
				}
				if (puts.size() > commitCount) {
					putList(table, puts);
					puts.clear();
				}
			}
			putList(table, puts);
		} catch (IOException e) {
			LogWriter.addLog("error", "put数据失败！！！{}", e.getMessage());
			e.printStackTrace();
		} finally {
			closeTable(table);
		}
	}

	public static void batchPut(HTable table, String rowKeyColumn,
			List<Map<String, Object>> rows) {

		List<Put> puts = new ArrayList<Put>();

		List<HbaseRowData> rowObjs = new ArrayList<HbaseRowData>();
		for (Map<String, Object> row : rows) {
			String[] rowKeyColumns = StringUtils
					.splitByWholeSeparatorPreserveAllTokens(rowKeyColumn, ",");
			Iterator<String> columnNames = row.keySet().iterator();
			StringBuilder builder = new StringBuilder();
			String rowKey = null;
			HbaseRowData rowObj = new HbaseRowData();
			while (columnNames.hasNext()) {
				String columnName = columnNames.next();
				for (String column : rowKeyColumns) {
					if (columnName.equalsIgnoreCase(column)) {
						builder.append(String.valueOf(row.get(columnName)));
					}
				}
				HbaseColumn columnObj = new HbaseColumn(columnName, columnName,
						String.valueOf(row.get(columnName)));
				rowObj.getHbaseColumns().add(columnObj);
			}
			rowKey = builder.toString();
			rowObj.setRowKey(rowKey);
			rowObjs.add(rowObj);

		}
		pushPuts(table, puts, rowObjs);

	}

	public static void pushPuts(HTable table, List<Put> puts,
			List<HbaseRowData> rowObjs) {
		for (HbaseRowData row : rowObjs) {
			Put put = new Put(Bytes.toBytes(row.getRowKey()));
			for (HbaseColumn column : row.getHbaseColumns()) {
				put.add(Bytes.toBytes(column.getColumnFamily()),
						Bytes.toBytes(column.getColumnName()),
						Bytes.toBytes(column.getColumnValue()));
			}
			puts.add(put);
		}

		putList(table, puts);
		puts.clear();
		rowObjs.clear();
	}

	public static void put(String tableName, String cf,
			List<Map<String, Object>> rows, String[] pkColumnNames) {
		HTable table = null;
		try {
			table = new HTable(getConf(), tableName);
			List<Put> puts = new ArrayList<Put>();
			int commitCount = 10000;// 每次最多提交的行数

			for (Map<String, Object> row : rows) {
				Iterator<String> columnNames = row.keySet().iterator();
				String rowKey = getRowKey(row, pkColumnNames);
				Put put = new Put(Bytes.toBytes(rowKey));
				while (columnNames.hasNext()) {
					String columnName = columnNames.next();
					String columnValue = String.valueOf(row.get(columnName));
					put.add(Bytes.toBytes(cf), Bytes.toBytes(columnName),
							Bytes.toBytes(columnValue));
					puts.add(put);
				}
				if (puts.size() > commitCount) {
					putList(table, puts);
					puts.clear();
				}
			}
			putList(table, puts);
		} catch (IOException e) {
			LogWriter.addLog("warn", "put数据失败！！！{}", e.getMessage());
		} finally {
			closeTable(table);
		}
	}

	public static Result get(String tableName, String rowKey) {
		HTable table = null;
		Result rs = null;
		try {
			table = new HTable(getConf(), tableName);
			Get get = new Get(Bytes.toBytes(rowKey));
			rs = table.get(get);
		} catch (IOException e) {
			LogWriter.addLog("warn", "获取行[" + rowKey + "]数据失败！！！{}",
					e.getMessage());
		} finally {
			closeTable(table);
		}
		return rs;
	}

	protected static void putList(HTable table, List<Put> puts) {
		if (CollectionUtils.isNotEmpty(puts)) {
			try {
				table.put(puts);
				table.flushCommits();
			} catch (IOException e) {
				LogWriter.addLog("error", "批量put数据失败！！！{}", e.getMessage());
			}
		}
	}

	public static void closeTable(HTable table) {
		if (table != null) {
			try {
				table.close();
			} catch (IOException e) {
				LogWriter.addLog("error", "关闭HTable失败！！！{}", e.getMessage());
			}
		}
	}

	public static void closeAdmin(HBaseAdmin admin) {
		if (admin != null) {
			try {
				admin.close();
			} catch (IOException e) {
				LogWriter
						.addLog("error", "关闭HBaseAdmin失败！！！{}", e.getMessage());
			}
		}
	}

	public static void scan(String tableName) {
		try {
			HTable table = new HTable(getConf(), tableName);
			Scan s = new Scan();
			SingleColumnValueFilter sf1 = new SingleColumnValueFilter(
					Bytes.toBytes("cf"), Bytes.toBytes("TIME_ID"),
					CompareOp.EQUAL, new BinaryComparator(
							Bytes.toBytes("201206")));
			SingleColumnValueFilter sf2 = new SingleColumnValueFilter(
					Bytes.toBytes("cf"), Bytes.toBytes("AREA_ID"),
					CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("730")));
			SingleColumnValueFilter sf3 = new SingleColumnValueFilter(
					Bytes.toBytes("cf"), Bytes.toBytes("SVC_BRND_ID"),
					CompareOp.EQUAL, new BinaryComparator(Bytes.toBytes("1")));
			FilterList filter = new FilterList(Operator.MUST_PASS_ALL, sf1,
					sf2, sf3);
			s.setFilter(filter);
			ResultScanner rs = table.getScanner(s);
			for (Result r : rs) {
				printResult(r);
			}
			rs.close();

		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public static void printResult(Result result) {
		for (KeyValue kv : result.raw()) {
			System.out.print(new String(kv.getRow()) + "\t");
			System.out.print(new String(kv.getFamily()) + ":");
			System.out.print(new String(kv.getQualifier()) + "\t");
			System.out.print(kv.getTimestamp() + "\t");
			System.out.println(new String(kv.getValue()));
		}
	}

	public static Configuration getConf() {
		return conf;
	}

	public static void setConf(Configuration conf) {
		HbaseUtil.conf = conf;
	}
}
